KAFKA-20287 : Fix CF handle leaks by muralibasani · Pull Request #21751 · apache/kafka

muralibasani · 2026-03-13T19:13:01Z

RocksDBStore.java — openRocksDB() base method: Added finally block after RocksDB.open() to close all CF handles and db if createColumnFamilies() or mergeColumnFamilyHandleLists() fails.
RocksDBStore.java — openDB(): Added finally block with closeNativeResources() to close all partially-initialized native resources if openRocksDB() or cfAccessor.open() fails.
RocksDBTimestampedStore.java: Changed RocksIterator from manual close() to try-with-resources, and added finally block to close all CF handles if an exception occurs before the accessor takes ownership.
RocksDBMigratingSessionStoreWithHeaders.java: Same as above — try-with-resources for RocksIterator and finally block for CF handle cleanup on failure.
RocksDBTimestampedStoreWithHeaders.java — openFromDefaultStore(): Added finally block to close all CF handles if an exception occurs before the accessor takes ownership.
RocksDBTimestampedStoreWithHeaders.java — openFromTimestampedStore(): Replaced manual per-handle close() calls (which missed columnFamilies.get(3)) with a single finally block that loops over all handles on failure.

Reviewers: Matthias J. Sax matthias@confluent.io

muralibasani · 2026-03-13T19:36:36Z

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java

-            throw new ProcessorStateException(fatalMessage, fatal);
-        } catch (final RocksDBException e) {
-            throw new ProcessorStateException("Error opening store " + name, e);
+            openRocksDB(dbOptions, columnFamilyOptions);


This method internally calls createColumnFamilies() or mergeColumnFamilyHandleLists() which might fail

I am wondering if we should do the necessary exception-handling inside openRocksDB instead?

I see openRocksDB() already handles its own cleanup (CF handles + db), but the closeNativeResources() in openDB() covers a different failure scope that openRocksDB() can't handle.

These 6 native resources (userSpecifiedOptions, cache, filter, wOptions, fOptions, statistics) are all created in openDB() before openRocksDB() is ever called and openRocksDB() doesn't own them and has no references to clean them up.

If openRocksDB() itself fails, it will cleanup everything which is in finally block.

If openRocksDB() does not fail but cfAccessor.open() fails — now db is open. Hence closeNativeResources handles full teardown.

Other option is to move everything into openRocksDB(), but this would change the method signature across all overrides.

muralibasani · 2026-03-13T19:44:55Z

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java

+            success = true;
+        } finally {
+            if (!success) {
+                closeNativeResources();


if openRocksDB() or cfAccessor.open() fail

muralibasani · 2026-03-13T19:49:30Z

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStore.java

-            noTimestampColumnFamily.close();
+        boolean success = false;
+        try {
+            try (final RocksIterator noTimestampsIter = db.newIterator(noTimestampColumnFamily)) {


updating with try-resources

muralibasani · 2026-03-13T19:50:04Z

...c/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java

+        boolean success = false;
+        try {
+            // Check if default CF has data (plain store upgrade)
+            try (final RocksIterator defaultIter = db.newIterator(defaultCf)) {


also updating with try-with resources

muralibasani · 2026-03-13T19:51:31Z

...c/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java

+        boolean success = false;
+        try {
+            // verify and close empty Default ColumnFamily
+            try (final RocksIterator defaultIter = db.newIterator(columnFamilies.get(0))) {


updating here as well in the same way try with resources

muralibasani · 2026-03-13T19:55:06Z

...n/java/org/apache/kafka/streams/state/internals/RocksDBMigratingSessionStoreWithHeaders.java

-            noHeadersColumnFamily.close();
+        boolean success = false;
+        try {
+            try (final RocksIterator noHeadersIter = db.newIterator(noHeadersColumnFamily)) {


updating here with try-with resources

muralibasani · 2026-03-16T15:20:17Z

@mjsax do you think this pr addresses the cf handle leaks ? Possible to review ?

mjsax

Thanks for the PR. Made a first pass. -- In the meantime, we added RocksDBMigratingWindowStoreWithHeaders that we should include in this PR.

mjsax · 2026-03-20T00:42:25Z

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java

+     * Used only by the error cleanup path in {@link #openDB} where some resources
+     * may not have been initialized yet.
+     */
+    private void closeNativeResources() {


We do already have code in close() which closes all these resources -- are these new helper necessary? I would believe that the runtime ensures that we call RocksDBStore.close() if init() (which calls openDB()) fails.

In close method, we are checking isOpen method which in turn checks open.get()

this would wait for openDB/openRocksDb() and if these fail, it would return false.

this 'false' does not close all resources in close() method.

mjsax · 2026-03-20T00:44:54Z

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java

-            throw new ProcessorStateException(fatalMessage, fatal);
-        } catch (final RocksDBException e) {
-            throw new ProcessorStateException("Error opening store " + name, e);
+            openRocksDB(dbOptions, columnFamilyOptions);


I am wondering if we should do the necessary exception-handling inside openRocksDB instead?

mjsax · 2026-03-20T00:49:47Z

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java

+                    mergeColumnFamilyHandleLists(existingColumnFamilies, createdColumnFamilies, allDescriptors);
+                openSuccess = true;
+                return result;
+            } finally {


Why are we not use catch(Exception e) -- this allows us to drop openSuccess flag?

Agree with removing success flag. Updated with catch block.

mjsax · 2026-03-20T00:50:37Z

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java

+                    for (final ColumnFamilyHandle handle : createdColumnFamilies) {
+                        handle.close();
+                    }
+                    db.close();


I don't think we need to close the db -- this should happen in RocksDBStore.close() already?

Yes, db.close() is not required again. Removed.

mjsax · 2026-03-20T00:51:26Z

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java

                    .collect(Collectors.toList());
            final List<ColumnFamilyHandle> existingColumnFamilies = new ArrayList<>(existingDescriptors.size());
+            final List<ColumnFamilyHandle> createdColumnFamilies = new ArrayList<>();
            db = RocksDB.open(dbOptions, absolutePath, existingDescriptors, existingColumnFamilies);


I think this call should also go inside the try-catch block?

Yes. The RocksDB.open() call is now inside the try-catch block, so if createColumnFamilies() or mergeColumnFamilyHandleLists() throws, the existingColumnFamilies handles populated by RocksDB.open() are properly cleaned up.

mjsax · 2026-03-20T00:53:22Z

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java

+            boolean openSuccess = false;
+            try {
+                createdColumnFamilies.addAll(db.createColumnFamilies(toCreate));
+                final List<ColumnFamilyHandle> result =


Suggested change

final List<ColumnFamilyHandle> result =

final List<ColumnFamilyHandle> allColumnFamilies =

Now returning it inline, instead of new var

mjsax · 2026-03-20T00:55:09Z

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStore.java

+                }
+            }
+            success = true;
+        } finally {


Why not use catch(Exception) and get rid of success flag?

Done. But also had to replace Exception with RunTimeException (spotbugs)

mjsax · 2026-03-20T00:55:35Z

...c/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java

+                }
+            }
+            success = true;
+        } finally {


mjsax · 2026-03-20T00:56:16Z

...c/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java

+                }
+            }
+            success = true;
+        } finally {


Applied the above change here too.

mjsax · 2026-03-20T00:57:38Z

...n/java/org/apache/kafka/streams/state/internals/RocksDBMigratingSessionStoreWithHeaders.java

+                }
+            }
+            success = true;
+        } finally {


We should use catch(Exception) instead

aliehsaeedii

Thanks @muralibasani. Good catch.
Just needs a rebase (and former comments from Matthias)

muralibasani · 2026-03-20T16:10:24Z

@mjsax @aliehsaeedii Updated PR based on comments.
Pls take another look.

muralibasani · 2026-03-20T16:57:26Z

...in/java/org/apache/kafka/streams/state/internals/RocksDBMigratingWindowStoreWithHeaders.java

-                noHeadersColumnFamily.close();
+        try {
+            // Check if DEFAULT CF has data (upgrade from old format without headers)
+            try (final RocksIterator noHeadersIter = db.newIterator(noHeadersColumnFamily)) {


moving newIterator() into try-resources

mjsax

Thanks. A few more cosmetics. Overall LGTM.

mjsax · 2026-03-20T22:14:25Z

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java

        setupStatistics(configs, dbOptions);
-        openRocksDB(dbOptions, columnFamilyOptions);
-        dbAccessor = new DirectDBAccessor(db, fOptions, wOptions);
+        boolean success = false;


Seems you missed this one for removal?

Ah yes. Done.

mjsax · 2026-03-20T22:14:46Z

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java

+                throw new ProcessorStateException("Error opening store " + name, e);
+            }
+            success = true;
+        } finally {


Should we use catch() instead?

Agree, removed.

mjsax · 2026-03-20T22:16:32Z

...n/java/org/apache/kafka/streams/state/internals/RocksDBMigratingSessionStoreWithHeaders.java

+                        withHeadersColumnFamily,
+                        HeadersBytesStore::convertToHeaderFormat,
+                        this,
+                            open


Suggested change

open

open

mjsax · 2026-03-20T22:18:53Z

streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStore.java

+                        noTimestampColumnFamily,
+                        withTimestampColumnFamily,
+                        TimestampedBytesStore::convertToTimestampedFormat,
+                        this, open


Suggested change

this, open

this,

open

mjsax · 2026-03-20T22:19:28Z

...c/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java

+                        headersCf,
+                        HeadersBytesStore::convertFromPlainToHeaderFormat,
+                        this,
+                            open


Suggested change

open

open

mjsax · 2026-03-20T22:21:12Z

...n/java/org/apache/kafka/streams/state/internals/RocksDBMigratingSessionStoreWithHeaders.java

-            log.info("Opening store {} in regular mode", name);
-            cfAccessor = new SingleColumnFamilyAccessor(offsetsCf, withHeadersColumnFamily);
-            noHeadersColumnFamily.close();
+        try {


Why do we need to nest two try-catch block? With success flag begin remove, isn't one sufficient? (Also elsewhere.)

Indeed not needed. Removed from others too.

muralibasani · 2026-03-21T17:34:08Z

Thanks. A few more cosmetics. Overall LGTM.

Thank you. Pls take another look.

github-actions bot added triage PRs from the community streams labels Mar 13, 2026

muralibasani commented Mar 13, 2026

View reviewed changes

muralibasani changed the title ~~Fix CF handle leaks~~ KAFKA-20287 : Fix CF handle leaks Mar 13, 2026

muralibasani marked this pull request as ready for review March 13, 2026 19:56

github-actions bot removed the triage PRs from the community label Mar 14, 2026

mjsax reviewed Mar 20, 2026

View reviewed changes

mjsax added kip Requires or implements a KIP ci-approved labels Mar 20, 2026

aliehsaeedii approved these changes Mar 20, 2026

View reviewed changes

muralibasani force-pushed the fix-cf-leaks branch from 0587c17 to 22b1dd2 Compare March 20, 2026 15:16

muralibasani commented Mar 20, 2026

View reviewed changes

mjsax reviewed Mar 20, 2026

View reviewed changes

muralibasani added 3 commits March 21, 2026 18:13

Fix CF handle leaks

9ac3fe9

Fix from reviews

1c031a5

Fix from reviews

230723a

muralibasani force-pushed the fix-cf-leaks branch from 2f24bc1 to 230723a Compare March 21, 2026 17:13

Remove double try blocks, indentation issues

76bd45d

muralibasani requested a review from mjsax March 21, 2026 17:32

	final List<ColumnFamilyHandle> result =
	final List<ColumnFamilyHandle> allColumnFamilies =

Conversation

muralibasani commented Mar 13, 2026 • edited by mjsax Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

muralibasani commented Mar 16, 2026

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aliehsaeedii left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

muralibasani commented Mar 20, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

muralibasani commented Mar 13, 2026 •

edited by mjsax

Loading

aliehsaeedii left a comment •

edited

Loading